package com.small.nacos.common.notify;


import com.alibaba.nacos.api.exception.runtime.NacosRuntimeException;
import com.alibaba.nacos.common.notify.SlowEvent;
import com.alibaba.nacos.common.utils.ClassUtils;
import com.small.nacos.common.notify.listener.Subscriber;
import com.small.nacos.common.utils.BiFunction;
import com.small.nacos.common.utils.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Map;
import java.util.ServiceLoader;
import java.util.concurrent.ConcurrentHashMap;

import static com.alibaba.nacos.api.exception.NacosException.SERVER_ERROR;


/**
 * @Author zhoujin
 * @Date 2022/11/7 13:56
 */
public class NotifyCenter {

    private static final Logger LOGGER = LoggerFactory.getLogger(NotifyCenter.class);

    private static final NotifyCenter INSTANCE = new NotifyCenter();

    public static int ringBufferSize = 16384;

    private static BiFunction<Class<? extends Event>, Integer, EventPublisher> publisherFactory = null;

    private static Class<? extends EventPublisher> clazz = null;

    private final Map<String, EventPublisher> publisherMap = new ConcurrentHashMap<String, EventPublisher>(16);

    static {

        final ServiceLoader<EventPublisher> loader = ServiceLoader.load(EventPublisher.class);
        Iterator<EventPublisher> iterator = loader.iterator();

        if (iterator.hasNext()) {
            clazz = iterator.next().getClass();
        } else {
            clazz = DefaultPublisher.class;
        }

        publisherFactory = new BiFunction<Class<? extends Event>, Integer, EventPublisher>() {

            @Override
            public EventPublisher apply(Class<? extends Event> cls, Integer buffer) {
                try {
                    EventPublisher publisher = clazz.newInstance();
                    publisher.init(cls, buffer);
                    return publisher;
                } catch (Throwable ex) {
                    LOGGER.error("Service class newInstance has error : {}", ex);
                    throw new NacosRuntimeException(SERVER_ERROR, ex);
                }
            }
        };

    }


    public static EventPublisher registerToPublisher(final Class<? extends Event> eventType, final int queueMaxSize) {

        final String topic = ClassUtils.getCanonicalName(eventType);
        synchronized (NotifyCenter.class) {
            // MapUtils.computeIfAbsent is a unsafe method.
            MapUtils.computeIfAbsent(INSTANCE.publisherMap, topic, publisherFactory, eventType, queueMaxSize);
        }
        return INSTANCE.publisherMap.get(topic);
    }


    public static <T> void registerSubscriber(final Subscriber consumer) {
        final Class<? extends Event> subscribeType = consumer.subscribeType();
        addSubscriber(consumer, subscribeType);
    }

    private static void addSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType) {

        final String topic = ClassUtils.getCanonicalName(subscribeType);
        synchronized (NotifyCenter.class) {
            // MapUtils.computeIfAbsent is a unsafe method.
            MapUtils.computeIfAbsent(INSTANCE.publisherMap, topic, publisherFactory, subscribeType, ringBufferSize);
        }
        EventPublisher publisher = INSTANCE.publisherMap.get(topic);
        publisher.addSubscriber(consumer);
    }

    public static boolean publishEvent(final Event event) {
        try {
            return publishEvent(event.getClass(), event);
        } catch (Throwable ex) {
            LOGGER.error("There was an exception to the message publishing : {}", ex);
            return false;
        }
    }

    private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) {

        final String topic = ClassUtils.getCanonicalName(eventType);

        EventPublisher publisher = INSTANCE.publisherMap.get(topic);
        if (publisher != null) {
            return publisher.publish(event);
        }
        LOGGER.warn("There are no [{}] publishers for this event, please register", topic);
        return false;
    }
}
